package com.rocketmq.demo;

import com.rocketmq.constants.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.junit.Test;

public class GTagTest {

    /**
     * 发送带标签的消息
     *
     * @throws Exception
     */
    @Test
    public void tagProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("tag-producer-group");
        producer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        producer.start();

        // 创建一个带tag的消息
        Message message1 = new Message("tagTopic", "vip1", "我是vip1的文章".getBytes());
        Message message2 = new Message("tagTopic", "vip2", "我是vip2的文章".getBytes());
        producer.send(message1);
        producer.send(message2);

        System.out.println("发送成功");
        producer.shutdown();
    }

    // 订阅关系一致性
    @Test
    public void tagConsumer1() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-a");
        consumer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        consumer.subscribe("tagTopic", "vip1");
        consumer.registerMessageListener((MessageListenerConcurrently) (messageExtList, context) -> {
            System.out.println("我是VIP1的消费者 消费消息：" + new String(messageExtList.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }

    @Test
    public void tagConsumer2() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tag-consumer-group-b");
        consumer.setNamesrvAddr(MqConstants.NAMESERVER_ADDR);
        consumer.subscribe("tagTopic", "vip1 || vip2");
        consumer.registerMessageListener((MessageListenerConcurrently) (messageExtList, context) -> {
            System.out.println("我是VIP2的消费者 消费消息：" + new String(messageExtList.get(0).getBody()));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        System.in.read();
    }
}
